import akshare as ak
import pandas as pd
from sqlalchemy import create_engine, text

# MySQL 连接配置
engine = create_engine('mysql+pymysql://root:123456@localhost:3306/cn_stock_data')

# 示例报告期
years = range(2020, 2026)
quarters = ["0331", "0630", "0930", "1231"]
report_dates = [f"{year}{q}" for year in years for q in quarters]

def process_performance_forecast(report_date_str: str):
    print(f"开始处理 {report_date_str} ...")
    try:
        df = ak.stock_yjyg_em(date=report_date_str)
        if df.empty:
            print(f"⚠️ {report_date_str} 没有数据，跳过")
            return

        # 删除序号列
        if '序号' in df.columns:
            df = df.drop(columns=['序号'])

        # 数字列处理
        numeric_cols = ['预测数值', '业绩变动幅度', '上年同期值']
        for col in numeric_cols:
            if col in df.columns:
                df[col] = pd.to_numeric(df[col], errors='coerce')

        # 日期列处理
        if '公告日期' in df.columns:
            df['公告日期'] = pd.to_datetime(df['公告日期'], errors='coerce').dt.date

        df['report_date'] = pd.to_datetime(report_date_str, format='%Y%m%d').date()

        # 重命名列与表字段对应
        df = df.rename(columns={
            '股票代码': 'stock_code',
            '股票简称': 'stock_name',
            '预测指标': 'forecast_item',
            '业绩变动': 'performance_change',
            '预测数值': 'forecast_value',
            '业绩变动幅度': 'performance_change_ratio',
            '业绩变动原因': 'performance_change_reason',
            '预告类型': 'forecast_type',
            '上年同期值': 'last_year_value',
            '公告日期': 'announcement_date'
        })

        df = df.drop_duplicates(
            subset=['stock_code', 'report_date', 'forecast_item'],
            keep='last'   # 或 'first'，看你想保留哪条
        )

        # 删除旧数据
        with engine.begin() as conn:
            conn.execute(
                text("DELETE FROM performance_forecast WHERE report_date = :report_date"),
                {"report_date": df['report_date'].iloc[0]}
            )

        # 插入前只保留表中已有列
        with engine.begin() as conn:
            table_cols = [row[0] for row in conn.execute(text("SHOW COLUMNS FROM performance_forecast"))]
        df_to_insert = df[[c for c in df.columns if c in table_cols]]

        # 插入数据
        df_to_insert.to_sql('performance_forecast', con=engine, if_exists='append', index=False)
        print(f"✅ {report_date_str} 导入完成！ 共 {len(df)} 行")

    except Exception as e:
        print(f"❌ {report_date_str} 处理失败: {e}")

# 批量处理
for date in report_dates:
    process_performance_forecast(date)

print("🎉 所有业绩预告数据导入完成！")
